import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.yeph.sources.BusinessDataSource;

public class FlinkFirst {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT, "8081");
        StreamExecutionEnvironment evn = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        evn.setParallelism(2);
        DataStreamSource<Tuple2<String, Integer>> business = evn.addSource(new BusinessDataSource());

        KeyedStream<Tuple2<String, Integer>, String> sum = business.keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0);
        sum.addSink(new SinkFunction<Tuple2<String, Integer>>() {
            @Override
            public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
                System.out.println(value);
            }
        });


        evn.execute("test");

    }
}
